import asyncio
import collections
import enum
import io
import itertools
import logging
import random
import struct
from typing import Any
from typing import Awaitable
from typing import Callable
from typing import cast
from typing import Deque
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
from typing import Union

Data = Union[str, bytes]

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s][%(filename)s:%(lineno)d] - %(message)s",
)
logger = logging.getLogger(__name__)


class opcode(enum.IntEnum):
    OP_BINARY = 1
    OP_PING = 2
    OP_PONG = 3
    OP_CONT = 4
    OP_CLOSE = 5


class Frame(object):
    def __init__(self, fin: bool, opcode: int, data: bytes):
        self.fin = fin
        self.opcode = opcode
        self.data = data

    def __repr__(self):
        return f"Frame(opcode={opcode(self.opcode).name}, data=[###], fin={self.fin})"

    @classmethod
    async def read(cls, read_fn: Callable[[int], Awaitable[bytes]]) -> "Frame":
        """
        读取一帧
        """
        # 读取两字节的header
        data = await read_fn(2)
        head1, head2 = struct.unpack("!BB", data)

        # NOTE: 这样比调用bool()稍微快些
        fin = True if head1 & 0b10000000 else False
        opcode = head1 & 0b00001111

        length = head2
        if length == 254:
            data = await read_fn(2)  # 读取16bit作为实际长度
            (length,) = struct.unpack("!H", data)
        elif length == 255:
            data = await read_fn(8)  # 额外读取64bit作为实际长度
            (length,) = struct.unpack("!Q", data)

        # 读取payload
        data = await read_fn(length)

        # 组装成frame
        frame = cls(fin, opcode, data)
        frame.check()

        return frame

    def write(self, write_fn: Callable[[bytes], Any]) -> None:
        """
        写入一帧
        """
        self.check()

        output = io.BytesIO()

        # 准备header
        head1 = (0b10000000 if self.fin else 0) | self.opcode

        length = len(self.data)
        if length < 254:
            output.write(struct.pack("!BB", head1, length))
        elif length < 65536:
            output.write(struct.pack("!BBH", head1, 254, length))
        else:
            assert length < 2**64
            output.write(struct.pack("!BBQ", head1, 255, length))

        # 准备data
        data = self.data
        output.write(data)

        # 写入帧

        # The frame is written in a single call to write in order to prevent
        # TCP fragmentation. See #68 for details. This also makes it safe to
        # send frames concurrently from multiple coroutines.
        write_fn(output.getvalue())

    def check(self) -> None:
        """
        Check that reserved bits and opcode have acceptable values.

        :raises ~websockets.exceptions.ProtocolError: if a reserved
            bit or the opcode is invalid

        """
        # The first parameter is called `frame` rather than `self`,
        # but it's the instance of class to which this method is bound.

        DATA_OPCODES = [opcode.OP_CONT, opcode.OP_BINARY]
        CTRL_OPCODES = [opcode.OP_CLOSE, opcode.OP_PING, opcode.OP_PONG]
        if self.opcode in DATA_OPCODES:
            return
        elif self.opcode in CTRL_OPCODES:
            if len(self.data) > 125:
                raise ProtocolError("control frame too long")
            if not self.fin:
                raise ProtocolError("fragmented control frame")
        else:
            raise ProtocolError(f"invalid opcode: {self.opcode}")


def prepare_data(data: Data) -> Tuple[int, bytes]:
    """用于data帧"""
    if isinstance(data, str):
        return opcode.OP_BINARY, data.encode()
    elif isinstance(data, (bytes, bytearray)):
        return opcode.OP_BINARY, data
    elif isinstance(data, memoryview):
        if data.c_contiguous:
            return opcode.OP_BINARY, data
        else:
            return opcode.OP_BINARY, data.tobytes()

    else:
        raise TypeError("data must be bytes-like or str")


def encode_data(data: Data) -> bytes:
    """用于ping/pong帧"""
    if isinstance(data, str):
        return data.encode()
    elif isinstance(data, (bytes, bytearray)):
        return bytes(data)
    elif isinstance(data, memoryview):
        return data.tobytes()

    else:
        raise TypeError("data must be bytes-like or str")


def serialize_close(code: int, reason: str) -> bytes:
    return struct.pack("!H", code) + reason.encode()


def parse_close(data: bytes) -> Tuple[int, str]:
    """
    解析close帧

    返回 (code, reason)
    """
    length = len(data)
    if length >= 2:
        (code,) = struct.unpack("!H", data[:2])
        reason = data[2:].decode("utf-8")
        return code, reason
    elif length == 0:
        return 1005, ""
    else:
        assert length == 1
        raise ProtocolError("close frame too short")


CLOSE_CODES = {
    1000: "OK",
    1001: "going away",
    1002: "protocol error",
    1003: "unsupported type",
    # 1004 is reserved
    1005: "no status code [internal]",
    1006: "connection closed abnormally [internal]",
    1007: "invalid data",
    1008: "policy violation",
    1009: "message too big",
    1010: "extension required",
    1011: "unexpected error",
    1015: "TLS failure [internal]",
}


def format_close(code: int, reason: str) -> str:
    """
    Display a human-readable version of the close code and reason.
    """
    if 3000 <= code < 4000:
        explanation = "registered"
    elif 4000 <= code < 5000:
        explanation = "private use"
    else:
        explanation = CLOSE_CODES.get(code, "unknown")
    result = f"code = {code} ({explanation}), "

    if reason:
        result += f"reason = {reason}"
    else:
        result += "no reason"

    return result


class State(enum.IntEnum):
    CONNECTING = 0
    OPEN = 1
    CLOSING = 2
    CLOSED = 3


# 异常
class ConnectionClosed(Exception):
    """
    Raised when trying to interact with a closed connection.

    Provides the connection close code and reason in its ``code`` and
    ``reason`` attributes respectively.

    """

    def __init__(self, code: int, reason: str) -> None:
        self.code = code
        self.reason = reason
        super().__init__(format_close(code, reason))


class ConnectionClosedError(ConnectionClosed):
    """
    Like :exc:`ConnectionClosed`, when the connection terminated with an error.

    This means the close code is different from 1000 (OK) and 1001 (going away).

    """

    def __init__(self, code: int, reason: str) -> None:
        assert code != 1000 and code != 1001
        super().__init__(code, reason)


class ConnectionClosedOK(ConnectionClosed):
    """
    Like :exc:`ConnectionClosed`, when the connection terminated properly.

    This means the close code is 1000 (OK) or 1001 (going away).

    """

    def __init__(self, code: int, reason: str) -> None:
        assert code == 1000 or code == 1001
        super().__init__(code, reason)


class InvalidState(Exception):
    """
    Raised when an operation is forbidden in the current state.

    This exception is an implementation detail.

    It should never be raised in normal circumstances.
    """


class ProtocolError(Exception):
    """
    Raised when the other side breaks the protocol.
    """


class Protocol(asyncio.Protocol):
    """数据传输协议"""

    side: str = "undefined"

    def __init__(self):
        self.loop = asyncio.get_running_loop()
        self.reader = asyncio.StreamReader(loop=self.loop)

        self.ping_interval = 10
        self.ping_timeout = 20
        self.close_timeout = 20
        self.max_queue = 32
        self.write_limit = 2**16
        self.legacy_recv = False

        # Copied from asyncio.FlowControlMixin
        self._paused = False
        self._drain_waiter: Optional[asyncio.Future[None]] = None
        self._drain_lock = asyncio.Lock()

        self.state = State.CONNECTING
        logger.debug(f"{self.side} - state = CONNECTING")

        self.close_code: int
        self.close_reason: str

        self.connection_lost_waiter: asyncio.Future[None] = self.loop.create_future()

        # 用于接收数据的队列
        self.messages: Deque[Data] = collections.deque()
        self._pop_message_waiter: Optional[asyncio.Future[None]] = None
        self._put_message_waiter: Optional[asyncio.Future[None]] = None

        # ping IDs 到 waiters 的映射, 时间顺序排序
        self.pings: Dict[bytes, asyncio.Future[None]] = {}

        # 数据传输任务
        self.transfer_data_task: asyncio.Task[None]
        # 数据传输过程中可能出现的异常
        self.transfer_data_exc: Optional[BaseException] = None
        # 发送keepalive ping的任务
        self.keepalive_ping_task: asyncio.Task[None]
        # 关闭tcp连接的任务
        self.close_connection_task: asyncio.Task[None]

    # Copied from asyncio.FlowControlMixin
    async def _drain_helper(self) -> None:  # pragma: no cover
        if self.connection_lost_waiter.done():
            raise ConnectionResetError("Connection lost")
        if not self._paused:
            return
        waiter = self._drain_waiter
        assert waiter is None or waiter.cancelled()
        waiter = self.loop.create_future()
        self._drain_waiter = waiter
        await waiter

    # Copied from asyncio.StreamWriter
    async def _drain(self) -> None:  # pragma: no cover
        if self.reader is not None:
            exc = self.reader.exception()
            if exc is not None:
                raise exc
        if self.transport is not None:
            if self.transport.is_closing():
                # Yield to the event loop so connection_lost() may be
                # called.  Without this, _drain_helper() would return
                # immediately, and code that calls
                #     write(...); yield from drain()
                # in a loop would never call connection_lost(), so it
                # would not see an error when the socket is closed.
                await asyncio.sleep(0)
        await self._drain_helper()

    def connection_open(self):
        """
        回调函数. 当握手完成,连接建立, 进入 OPEN 状态, 开始进入数据传输阶段
        """
        assert self.state is State.CONNECTING
        # State: CONNECTING --> OPEN
        self.state = State.OPEN
        logger.debug(f"{self.side} - state = OPEN")

        # 开始执行任务
        self.transfer_data_task = self.loop.create_task(self.transfer_data())
        self.keepalive_ping_task = self.loop.create_task(self.keepalive_ping())
        self.close_connection_task = self.loop.create_task(self.close_connection())

    async def handshake(self):
        """
        开始握手
        """
        self.connection_open()

    async def wait_closed(self):
        """
        等待, 直到连接关闭
        """
        await asyncio.shield(self.connection_lost_waiter)

    async def send(self, message: Data):
        """
        发送消息
        """
        await self.ensure_open()

        opcode, data = prepare_data(message)
        fin = True
        await self.write_frame(fin, opcode, data)

    async def recv(self) -> Data:
        """
        接收下一个消息
        """
        if self._pop_message_waiter is not None:
            raise RuntimeError("另一个协程正在执行recv")

        # 等待队列中的消息或者连接断开
        while len(self.messages) <= 0:
            pop_message_waiter: asyncio.Future[None] = self.loop.create_future()
            self._pop_message_waiter = pop_message_waiter
            try:
                # If asyncio.wait() is canceled, it doesn't cancel
                # pop_message_waiter and self.transfer_data_task.
                await asyncio.wait(
                    [pop_message_waiter, self.transfer_data_task],
                    return_when=asyncio.FIRST_COMPLETED,
                )
            finally:
                self._pop_message_waiter = None

            # If asyncio.wait(...) exited because self.transfer_data_task
            # completed before receiving a new message, raise a suitable
            # exception (or return None if legacy_recv is enabled).
            if not pop_message_waiter.done():
                if self.legacy_recv:
                    return None  # type: ignore
                else:
                    # Wait until the connection is closed to raise
                    # ConnectionClosed with the correct code and reason.
                    await self.ensure_open()

        # Pop a message from the queue.
        message = self.messages.popleft()

        # 通知 transfer_data()
        if self._put_message_waiter is not None:
            self._put_message_waiter.set_result(None)
            self._put_message_waiter = None

        return message

    async def close(self, code: int = 1000, reason: str = ""):
        """
        结束握手
        """
        try:
            await asyncio.wait_for(
                self.write_close_frame(serialize_close(code, reason)),
                self.close_timeout,
            )
        except asyncio.TimeoutError:
            # 如果send buffer已满, close帧无法被发送, 关闭握手不可能完成.
            # 这种情况下 fail connection更快地关闭连接
            self.fail_connection()

        # 如果在超时时间内, 没有close帧接收到, 等待关闭transfer_data_task任务

        # 如果close()被调用多次, 其中一个超时, transfer_data_task任务将会取消,
        # 其他的调用将会收到CancelledError异常

        try:
            await asyncio.wait_for(self.transfer_data_task, self.close_timeout)
        except (asyncio.TimeoutError, asyncio.CancelledError):
            pass

        # 等待close_connection_task关闭tcp连接
        await asyncio.shield(self.close_connection_task)

    async def ping(self, data: Optional[Data] = None) -> Awaitable[None]:
        """
        发送一个ping
        """
        await self.ensure_open()

        if data is not None:
            data = encode_data(data)

        # 如果没有提供data, 生成随机数据
        while data is None or data in self.pings:
            data = struct.pack("!I", random.getrandbits(32))

        assert data not in self.pings
        self.pings[data] = self.loop.create_future()

        fin = True
        await self.write_frame(fin, opcode.OP_PING, data)

        return asyncio.shield(self.pings[data])

    async def pong(self, data: Data = b"") -> None:
        """
        发送一个pong
        """
        await self.ensure_open()

        data = encode_data(data)

        fin = True
        await self.write_frame(fin, opcode.OP_PONG, data)

    ##############################################
    # 下面是私有的方法

    async def ensure_open(self) -> None:
        """
        Check that the WebSocket connection is open.

        Raise :exc:`~websockets.exceptions.ConnectionClosed` if it isn't.

        """
        # 最可能出现的case放最前面
        if self.state is State.OPEN:
            # 如果transfer_data_task任务退出且没有进行关闭握手,
            # 等待close_connection_task任务关闭连接
            # 连接状态 OPEN --> CLOSED
            if self.transfer_data_task.done():
                await asyncio.shield(self.close_connection_task)
                raise self.connection_closed_exc()
            else:
                return

        if self.state is State.CLOSED:
            raise self.connection_closed_exc()

        if self.state is State.CLOSING:
            # 开始了关闭握手, 等待close_connection_task任务完成, 并获取code和reason
            await asyncio.shield(self.close_connection_task)
            raise self.connection_closed_exc()

        # 一般不可能到达
        assert self.state is State.CONNECTING
        raise InvalidState("WebSocket connection isn't established yet")

    def connection_closed_exc(self) -> ConnectionClosed:
        exception: ConnectionClosed
        if self.close_code == 1000 or self.close_code == 1001:
            exception = ConnectionClosedOK(self.close_code, self.close_reason)
        else:
            exception = ConnectionClosedError(self.close_code, self.close_reason)
        # Chain to the exception that terminated data transfer, if any.
        exception.__cause__ = self.transfer_data_exc
        return exception

    async def transfer_data(self) -> None:
        """
        读取到来的消息, 放到队列中
        协程对应的协程任务一直运行, 直到closing handshake开始
        """
        try:
            while True:
                message = await self.read_message()

                # message为None表示收到了一个close帧
                if message is None:
                    break

                # 等待直到队列中有存放空间
                if self.max_queue is not None:
                    while len(self.messages) >= self.max_queue:
                        self._put_message_waiter = self.loop.create_future()
                        try:
                            await asyncio.shield(self._put_message_waiter)
                        finally:
                            self._put_message_waiter = None

                # 消息存放到队列
                self.messages.append(message)

                # 通知 recv()
                if self._pop_message_waiter is not None:
                    self._pop_message_waiter.set_result(None)
                    self._pop_message_waiter = None

        except asyncio.CancelledError as exc:
            self.transfer_data_exc = exc
            raise

        except ProtocolError as exc:
            self.transfer_data_exc = exc
            self.fail_connection(1002)

        except (ConnectionError, EOFError) as exc:
            # Reading data with self.reader.readexactly may raise:
            # - most subclasses of ConnectionError if the TCP connection
            #   breaks, is reset, or is aborted;
            # - IncompleteReadError, a subclass of EOFError, if fewer
            #   bytes are available than requested.
            self.transfer_data_exc = exc
            self.fail_connection(1006)

        except UnicodeDecodeError as exc:
            self.transfer_data_exc = exc
            self.fail_connection(1007)

        except Exception as exc:
            # This shouldn't happen often because exceptions expected under
            # regular circumstances are handled above. If it does, consider
            # catching and handling more exceptions.
            logger.error("Error in data transfer", exc_info=True)

            self.transfer_data_exc = exc
            self.fail_connection(1011)

    async def read_message(self) -> Optional[Data]:
        """
        读取一个(完整的)消息

        一个完整的消息可能由多个data帧中的payload组成
        如果消息被分片了, 会自动重组
        如果closing handshake开始, 返回None
        """
        frame = await self.read_data_frame()

        # A close frame was received.
        if frame is None:
            return None

        assert frame.opcode == opcode.OP_BINARY
        # 最常见的情况: 没有分片, 第一个data帧就包含FIN标志位
        if frame.fin:
            return frame.data

        # 包含分片的情况:
        chunks: List[Data] = []

        def append(frame: Frame) -> None:
            chunks.append(frame.data)

        append(frame)

        # 直到还没读到FIN标志位就一直往下读下一个data帧
        while not frame.fin:
            frame = await self.read_data_frame()
            if frame is None:
                raise ProtocolError("incomplete fragmented message")
            # 正常情况下分片帧opcode应该是OP_CONT
            if frame.opcode != opcode.OP_CONT:
                raise ProtocolError("unexpected opcode")
            append(frame)

        return b"".join(chunks)  # type: ignore

    async def read_data_frame(self) -> Optional[Frame]:
        """
        读取一个data帧

        如果先出现control帧, 处理完继续循环等待, 直到出现data帧
        如果出现close帧, 返回None

        """
        # 6.2. Receiving Data
        while True:
            frame = await self.read_frame()

            # 5.5. Control Frames
            if frame.opcode == opcode.OP_CLOSE:
                # 7.1.5.  The WebSocket Connection Close Code
                # 7.1.6.  The WebSocket Connection Close Reason
                self.close_code, self.close_reason = parse_close(frame.data)
                try:
                    await self.write_close_frame(frame.data)
                except ConnectionClosed:
                    # It doesn't really matter if the connection was closed
                    # before we could send back a close frame.
                    pass
                return None

            elif frame.opcode == opcode.OP_PING:
                # Answer pings.
                ping_hex = frame.data.hex() or "[empty]"
                logger.debug(f"{self.side} - received ping, sending pong: {ping_hex}")
                await self.pong(frame.data)

            elif frame.opcode == opcode.OP_PONG:
                # Acknowledge pings on solicited pongs.
                pong_hex = frame.data.hex() or "[empty]"
                if frame.data in self.pings:
                    logger.debug(f"{self.side} - received solicited pong: {pong_hex}")
                    # Acknowledge all pings up to the one matching this pong.
                    ping_id = None
                    ping_ids = []
                    for ping_id, ping in self.pings.items():
                        ping_ids.append(ping_id)
                        if not ping.done():
                            ping.set_result(None)
                        if ping_id == frame.data:
                            break
                    else:  # pragma: no cover
                        assert False, "ping_id is in self.pings"
                    # Remove acknowledged pings from self.pings.
                    for ping_id in ping_ids:
                        del self.pings[ping_id]
                    ping_ids = ping_ids[:-1]
                    if ping_ids:
                        pings_hex = ", ".join(
                            ping_id.hex() or "[empty]" for ping_id in ping_ids
                        )
                        plural = "s" if len(ping_ids) > 1 else ""
                        logger.debug(
                            f"{self.side} - acknowledged previous ping{plural}: {pings_hex}"
                        )
                else:
                    pong_hex = (frame.data.hex() or "[empty]",)
                    logger.debug(f"{self.side} - received unsolicited pong: {pong_hex}")

            # 5.6. Data Frames
            else:
                return frame

    async def read_frame(self) -> Frame:
        """
        读取一个帧
        """
        frame = await Frame.read(self.reader.readexactly)
        logger.debug("%s < %r", self.side, frame)
        return frame

    async def write_frame(
        self, fin: bool, opcode: int, data: bytes, *, _expected_state: int = State.OPEN
    ) -> None:
        # 防御性断言, 协议遵从性检查
        if self.state is not _expected_state:  # pragma: no cover
            raise InvalidState(
                f"Cannot write to a WebSocket in the {self.state.name} state"
            )

        frame = Frame(fin, opcode, data)
        logger.debug("%s > %r", self.side, frame)
        frame.write(self.transport.write)

        try:
            # drain() cannot be called concurrently by multiple coroutines:
            # http://bugs.python.org/issue29930. Remove this lock when no
            # version of Python where this bugs exists is supported anymore.
            async with self._drain_lock:
                # Handle flow control automatically.
                await self._drain()
        except ConnectionError:
            # Terminate the connection if the socket died.
            self.fail_connection()
            # Wait until the connection is closed to raise ConnectionClosed
            # with the correct code and reason.
            await self.ensure_open()

    async def write_close_frame(self, data: bytes = b"") -> None:
        """
        写入close帧
        """
        # Test and set the connection state before sending the close frame to
        # avoid sending two frames in case of concurrent calls.
        if self.state is State.OPEN:
            # 7.1.3. The WebSocket Closing Handshake is Started
            self.state = State.CLOSING
            logger.debug("%s - state = CLOSING", self.side)

            # 7.1.2. Start the WebSocket Closing Handshake
            fin = True
            await self.write_frame(
                fin, opcode.OP_CLOSE, data, _expected_state=State.CLOSING
            )

    async def keepalive_ping(self) -> None:
        """
        定期发送一个ping帧并等待一个pong帧
        """
        if self.ping_interval is None:
            return

        try:
            while True:
                await asyncio.sleep(self.ping_interval)

                # ping() raises CancelledError if the connection is closed,
                # when close_connection() cancels self.keepalive_ping_task.

                # ping() raises ConnectionClosed if the connection is lost,
                # when connection_lost() calls abort_pings().

                ping_waiter = await self.ping()

                if self.ping_timeout is not None:
                    try:
                        await asyncio.wait_for(ping_waiter, self.ping_timeout)
                    except asyncio.TimeoutError:
                        logger.debug(f"{self.side} ! timed out waiting for pong")
                        self.fail_connection(1011)
                        break

        except asyncio.CancelledError:
            raise

        except ConnectionClosed:
            pass

        except Exception:
            logger.warning("Unexpected exception in keepalive ping task", exc_info=True)

    async def close_connection(self) -> None:
        """
        7.1.1. Close the WebSocket Connection

        When the opening handshake succeeds, :meth:`connection_open` starts
        this coroutine in a task. It waits for the data transfer phase to
        complete then it closes the TCP connection cleanly.

        When the opening handshake fails, :meth:`fail_connection` does the
        same. There's no data transfer phase in that case.

        """
        try:
            # Wait for the data transfer phase to complete.
            if hasattr(self, "transfer_data_task"):
                try:
                    await self.transfer_data_task
                except asyncio.CancelledError:
                    pass

            # Cancel the keepalive ping task.
            if hasattr(self, "keepalive_ping_task"):
                self.keepalive_ping_task.cancel()

            # A client should wait for a TCP close from the server.
            if self.side == "client" and hasattr(self, "transfer_data_task"):
                if await self.wait_for_connection_lost():
                    return
                logger.debug(f"{self.side} ! timed out waiting for TCP close")

            # Half-close the TCP connection if possible (when there's no TLS).
            if self.transport.can_write_eof():
                logger.debug(f"{self.side} x half-closing TCP connection")
                self.transport.write_eof()

                if await self.wait_for_connection_lost():
                    return
                logger.debug(f"{self.side} ! timed out waiting for TCP close")

        finally:
            # The try/finally ensures that the transport never remains open,
            # even if this coroutine is canceled (for example).

            # If connection_lost() was called, the TCP connection is closed.
            # However, if TLS is enabled, the transport still needs closing.
            # Else asyncio complains: ResourceWarning: unclosed transport.
            if self.connection_lost_waiter.done() and self.transport.is_closing():
                return

            # Close the TCP connection. Buffers are flushed asynchronously.
            logger.debug(f"{self.side} x closing TCP connection")
            self.transport.close()

            if await self.wait_for_connection_lost():
                return
            logger.debug(f"{self.side} ! timed out waiting for TCP close")

            # Abort the TCP connection. Buffers are discarded.
            logger.debug(f"{self.side} x aborting TCP connection")
            self.transport.abort()

            # connection_lost() is called quickly after aborting.
            await self.wait_for_connection_lost()

    async def wait_for_connection_lost(self) -> bool:
        """
        等待tcp连接关闭或者close_timeout超时

        Return ``True`` if the connection is closed and ``False`` otherwise.

        """
        if not self.connection_lost_waiter.done():
            try:
                await asyncio.wait_for(
                    asyncio.shield(self.connection_lost_waiter), self.close_timeout
                )
            except asyncio.TimeoutError:
                pass
        # Re-check self.connection_lost_waiter.done() synchronously because
        # connection_lost() could run between the moment the timeout occurs
        # and the moment this coroutine resumes running.
        return self.connection_lost_waiter.done()

    def fail_connection(self, code: int = 1006, reason: str = "") -> None:
        """
        7.1.7. Fail the WebSocket Connection

        This requires:

        1. Stopping all processing of incoming data, which means cancelling
           :attr:`transfer_data_task`. The close code will be 1006 unless a
           close frame was received earlier.

        2. Sending a close frame with an appropriate code if the opening
           handshake succeeded and the other side is likely to process it.

        3. Closing the connection. :meth:`close_connection` takes care of
           this once :attr:`transfer_data_task` exits after being canceled.

        (The specification describes these steps in the opposite order.)

        """
        logger.debug(
            f"{self.side} ! failing {self.state.name} WebSocket connection with code {code}"
        )

        # Cancel transfer_data_task if the opening handshake succeeded.
        # cancel() is idempotent and ignored if the task is done already.
        if hasattr(self, "transfer_data_task"):
            self.transfer_data_task.cancel()

        # Send a close frame when the state is OPEN (a close frame was already
        # sent if it's CLOSING), except when failing the connection because of
        # an error reading from or writing to the network.
        # Don't send a close frame if the connection is broken.
        if code != 1006 and self.state is State.OPEN:
            frame_data = serialize_close(code, reason)

            # Write the close frame without draining the write buffer.

            # Keeping fail_connection() synchronous guarantees it can't
            # get stuck and simplifies the implementation of the callers.
            # Not drainig the write buffer is acceptable in this context.

            # This duplicates a few lines of code from write_close_frame()
            # and write_frame().

            self.state = State.CLOSING
            logger.debug("%s - state = CLOSING", self.side)

            frame = Frame(True, opcode.OP_CLOSE, frame_data)
            logger.debug("%s > %r", self.side, frame)
            frame.write(self.transport.write)

        # Start close_connection_task if the opening handshake didn't succeed.
        if not hasattr(self, "close_connection_task"):
            self.close_connection_task = self.loop.create_task(self.close_connection())

    def abort_pings(self) -> None:
        """
        Raise ConnectionClosed in pending keepalive pings.

        They'll never receive a pong once the connection is closed.

        """
        assert self.state is State.CLOSED
        exc = self.connection_closed_exc()

        for ping in self.pings.values():
            ping.set_exception(exc)
            # If the exception is never retrieved, it will be logged when ping
            # is garbage-collected. This is confusing for users.
            # Given that ping is done (with an exception), canceling it does
            # nothing, but it prevents logging the exception.
            ping.cancel()

        if self.pings:
            pings_hex = ", ".join(ping_id.hex() or "[empty]" for ping_id in self.pings)
            plural = "s" if len(self.pings) > 1 else ""
            logger.debug(
                "%s - aborted pending ping%s: %s", self.side, plural, pings_hex
            )

    #############################################
    # asyncio.Protocol 方法

    # Base Protocol
    # Connection回调函数: 在连接建立时调用
    def connection_made(self, transport: asyncio.BaseTransport) -> None:
        """
        Configure write buffer limits.

        The high-water limit is defined by ``self.write_limit``.

        The low-water limit currently defaults to ``self.write_limit // 4`` in
        :meth:`~asyncio.WriteTransport.set_write_buffer_limits`, which should
        be all right for reasonable use cases of this library.

        This is the earliest point where we can get hold of the transport,
        which means it's the best point for configuring it.

        """
        logger.debug("%s - event = connection_made(%s)", self.side, transport)

        transport = cast(asyncio.Transport, transport)
        transport.set_write_buffer_limits(self.write_limit)
        self.transport = transport

        # Copied from asyncio.StreamReaderProtocol
        self.reader.set_transport(transport)

    # Base Protocol
    # Connection回调函数: 在连接lost或者closed时调用
    def connection_lost(self, exc: Optional[Exception]) -> None:
        """
        7.1.4. The WebSocket Connection is Closed.

        """
        logger.debug("%s - event = connection_lost(%s)", self.side, exc)
        self.state = State.CLOSED
        logger.debug("%s - state = CLOSED", self.side)
        if not hasattr(self, "close_code"):
            self.close_code = 1006
        if not hasattr(self, "close_reason"):
            self.close_reason = ""
        logger.debug(
            "%s x code = %d, reason = %s",
            self.side,
            self.close_code,
            self.close_reason or "[no reason]",
        )
        self.abort_pings()
        # If self.connection_lost_waiter isn't pending, that's a bug, because:
        # - it's set only here in connection_lost() which is called only once;
        # - it must never be canceled.
        self.connection_lost_waiter.set_result(None)

        if True:  # pragma: no cover
            # Copied from asyncio.StreamReaderProtocol
            if self.reader is not None:
                if exc is None:
                    self.reader.feed_eof()
                else:
                    self.reader.set_exception(exc)

            # Copied from asyncio.FlowControlMixin
            # Wake up the writer if currently paused.
            if not self._paused:
                return
            waiter = self._drain_waiter
            if waiter is None:
                return
            self._drain_waiter = None
            if waiter.done():
                return
            if exc is None:
                waiter.set_result(None)
            else:
                waiter.set_exception(exc)

    # Base Protocol
    # Flow Control回调函数: 当传输缓冲区超过高水位时调用
    def pause_writing(self) -> None:  # pragma: no cover
        assert not self._paused
        self._paused = True

    # Base Protocol
    # Flow Control回调函数: 当传输的缓冲区低于低水位时调用
    def resume_writing(self) -> None:  # pragma: no cover
        assert self._paused
        self._paused = False

        waiter = self._drain_waiter
        if waiter is not None:
            self._drain_waiter = None
            if not waiter.done():
                waiter.set_result(None)

    # Streaming Protocol
    # 当接收到某些数据时调用, data是一个包含传入数据的非空字节对象
    def data_received(self, data: bytes) -> None:
        logger.debug("%s - event = data_received(<%d bytes>)", self.side, len(data))
        self.reader.feed_data(data)

    # Streaming Protocol
    # 当另一端发出不再发送任何数据的信号时调用
    # (例如，如果另一端也使用asyncio，则通过调用transport.write_eof())
    def eof_received(self) -> None:
        """
        Close the transport after receiving EOF.

        The WebSocket protocol has its own closing handshake: endpoints close
        the TCP or TLS connection after sending and receiving a close frame.

        As a consequence, they never need to write after receiving EOF, so
        there's no reason to keep the transport open by returning ``True``.

        Besides, that doesn't work on TLS connections.

        """
        logger.debug("%s - event = eof_received()", self.side)
        self.reader.feed_eof()
